//------------------------------------------------------------------------------
// <copyright file="RpcMessageDispatcher.cpp" company="Microsoft">
//     Copyright (c) Microsoft Corporation.  All rights reserved.
// </copyright>
//------------------------------------------------------------------------------

#include "stdafx.h"

#include "RpcMessageDispatcher.h"
#include "IRpcMessageHandler.h"
#include "IRpcChannel.h"
#include "RpcSocketTransportMessage.h"
#include "RpcSocketChannel.h"
#include "RpcChannelListener.h"
#include "RpcVersion.h"

using namespace Microsoft::Internal::GamesTest::Rpc::Server;

RpcMessageDispatcher::MessageHandlerTable RpcMessageDispatcher::s_activeMessageHandlers;
concurrency::critical_section RpcMessageDispatcher::s_socketListenerLock;

GAMESTEST_ANNOTATE_HRESULT_METHOD HRESULT RpcMessageDispatcher::RegisterMessageHandler(const GUID& serverId, int port, IRpcMessageHandler* messageHandler)
{
	// If you create two RpcServer objects on different threads at the same time,
	// then it is possible that they will enter this method at the same time, thus the lock.
	concurrency::critical_section::scoped_lock lock(s_socketListenerLock);

	HRESULT hr = S_OK;

	MessageHandlerTable::iterator iter = s_activeMessageHandlers.find(serverId);
	if(iter != s_activeMessageHandlers.end())
	{
		return GAMESTEST_RPC_E_SERVER_ID_ALREADY_IN_USE;
	}

	if(s_activeMessageHandlers.empty())
	{
		CHK(RpcChannelListener::Instance()->Start(port, &RpcMessageDispatcher::OnIncomingMessageReceived, &RpcMessageDispatcher::OnClientDisconnected));
	}

	s_activeMessageHandlers[serverId] = messageHandler;

end:
	return hr;
}

GAMESTEST_ANNOTATE_HRESULT_METHOD HRESULT RpcMessageDispatcher::UnregisterMessageHandler(const GUID& serverId)
{
	concurrency::critical_section::scoped_lock lock(s_socketListenerLock);

	HRESULT hr = S_OK;

	MessageHandlerTable::iterator iter = s_activeMessageHandlers.find(serverId);
	if(iter != s_activeMessageHandlers.end())
	{
		s_activeMessageHandlers.erase(iter);
	}

	if(s_activeMessageHandlers.empty())
	{
		CHK(RpcChannelListener::Instance()->Stop());
	}

end:
	return hr;
}

void RpcMessageDispatcher::OnIncomingMessageReceived(std::weak_ptr<IRpcChannel> client, RpcSocketTransportMessage socketTransportMessage)
{
	concurrency::critical_section::scoped_lock lock(s_socketListenerLock);

	HRESULT hr = S_OK;
	MessageHandlerTable::iterator iter = s_activeMessageHandlers.end();

	if (socketTransportMessage.version != GAMESTEST_RPC_VERSION)
	{
		hr = GAMESTEST_RPC_E_VERSION_MISMATCH;
		goto end;
	}

	iter = s_activeMessageHandlers.find(socketTransportMessage.serverId);

	if(iter != s_activeMessageHandlers.end())
	{
		IRpcMessageHandler* messageHandler = iter->second;

		CHK(messageHandler->ProcessIncomingMessage(client, socketTransportMessage.rpcMessage));
	}
	else
	{
		hr = GAMESTEST_RPC_E_UNKNOWN_SERVER_ID;
		goto end;
	}

end:
	if(FAILED(hr))
	{
#pragma warning(suppress: 6221)
		SendRpcMessageFailedMessage(socketTransportMessage.rpcMessage, socketTransportMessage, hr, client);
	}
}

void RpcMessageDispatcher::OnClientDisconnected(std::weak_ptr<IRpcChannel> client)
{
	for(auto iter = s_activeMessageHandlers.begin(); iter != s_activeMessageHandlers.end(); iter++)
	{
		(*iter).second->ClientDisconnected(client);
	}
}

void RpcMessageDispatcher::SendRpcMessageFailedMessage(const std::wstring& rpcMessage, RpcSocketTransportMessage socketTransportMessage, HRESULT errorCode, std::weak_ptr<IRpcChannel> client)
{
	HRESULT hr = S_OK;

	OutputDebugString(L"TDK RPC: Failed to handle incoming RPC message.  Request Content: ");
	OutputDebugString(rpcMessage.c_str());
	OutputDebugString(L"\n");
	RpcPrintError(errorCode);

	std::shared_ptr<IRpcChannel> sharedClient = client.lock();
	if(sharedClient != nullptr)
	{
		hr = sharedClient->SendRpcMessage(L"", socketTransportMessage.serverId, errorCode);
	}
	else
	{
		hr = GAMESTEST_RPC_E_CLIENT_DISCONNECTED;
	}

	if(FAILED(hr))
	{
		OutputDebugString(L"TDK RPC: An error ocurred in the transport and it was unable to send a response to the client.\n");
		RpcPrintError(hr);
	}
}

void RpcMessageDispatcher::Reset()
{
	s_activeMessageHandlers.clear();
}